package com.raylu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.raylu.realtime.bean.OrderDetail;
import com.raylu.realtime.bean.OrderInfo;
import com.raylu.realtime.bean.OrderWide;
import com.raylu.realtime.utils.DimUtil;
import com.raylu.realtime.utils.KafkaSourceUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.Period;
import java.util.Properties;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2021/12/30.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        Properties load = PropertiesUtil.load("config.properties");
        //TODO 1. 准备环境变量
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2. 设置检查点
        env.enableCheckpointing(5000L);
        //TODO 2.1. 设置检查点分解线对齐
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //TODO 2.2. 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
        //TODO 2.3. 设置当job cancel时是否保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //TODO 2.4. 设置失败重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        //TODO 2.5. 设置检查点存放位置
        env.setStateBackend(new FsStateBackend(load.getProperty("order.wide.app.fsstatebackend.url")));

        System.setProperty("HADOOP_USER_NAME", "raylu");
        //TODO 3.  从Kafka（DWD_ORDER_INFO）中读取OrderInfo数据
        KeyedStream<OrderInfo, String> orderInfoKeyedStream = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("order.wide.app.kafka.source-topic1"), load.getProperty("order.wide.app.kafka.group-id")))
                .map(new RichMapFunction<String, OrderInfo>() {
                    private SimpleDateFormat simpleDateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderInfo map(String value) throws Exception {
                        OrderInfo orderInfo = JSON.parseObject(value, OrderInfo.class);
                        orderInfo.setTs(simpleDateFormat.parse(orderInfo.getCreate_time()).getTime());
                        return orderInfo;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                            @Override
                            public long extractTimestamp(OrderInfo element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })).keyBy(r -> r.getId());

        //TODO 4. 从Kafka（DWD_ORDER_DETAIL）中读取OrderDetail数据
        KeyedStream<OrderDetail, String> orderDetailKeyedStream = env
                .addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("order.wide.app.kafka.source-topic2"), load.getProperty("order.wide.app.kafka.group-id")))
                .map(new RichMapFunction<String, OrderDetail>() {
                    private SimpleDateFormat simpleDateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderDetail map(String value) throws Exception {
                        OrderDetail orderDetail = JSON.parseObject(value, OrderDetail.class);
                        orderDetail.setTs(simpleDateFormat.parse(orderDetail.getCreate_time()).getTime());
                        return orderDetail;
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
                            @Override
                            public long extractTimestamp(OrderDetail element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })).keyBy(r -> r.getOrder_id());

        //TODO 5. 通过Interval Join 将两条流进行关联，OrderInfo流为主流设置间隔读取OrderDetail流
        SingleOutputStreamOperator<OrderWide> orderWideStreamOperator = orderInfoKeyedStream
                .intervalJoin(orderDetailKeyedStream)
                .between(Time.seconds(-5L), Time.seconds(5L))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {

                    @Override
                    public void processElement(OrderInfo left, OrderDetail right, ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>.Context ctx, Collector<OrderWide> out) throws Exception {
                        OrderWide orderWide = new OrderWide(left, right);
                        out.collect(orderWide);
                    }
                });
        //TODO 6. 通过map将OrderWide的数据进行维度补充
        orderWideStreamOperator.map(new RichMapFunction<OrderWide, OrderWide>() {
                    @Override
                    public OrderWide map(OrderWide value) throws Exception {
                        JSONObject jsonObject = DimUtil.qurayTableByPkWithCache(load.getProperty("base.db.app.cdc.phoenix.schem") + ".dim_user_info", Tuple2.of("id", value.getUser_id()));
                        String login_name = jsonObject.getString("LOGIN_NAME");
                        String name = jsonObject.getString("NAME");
                        String birthday = jsonObject.getString("BIRTHDAY");
                        String gender = jsonObject.getString("GENDER");
                        value.setLogin_name(login_name);
                        value.setName(name);
                        value.setBirthday(birthday);
                        value.setGender(gender);
                        LocalDate birthdayDate = LocalDate.parse(birthday);
                        Period between = Period.between(LocalDate.now(), birthdayDate);
                        int age = between.getYears();
                        value.setAge(age);
                        return value;
                    }
                }).print();


        env.execute();


    }
}
